package flink_p1

import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.redis.RedisSink
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig
import org.apache.flink.streaming.connectors.redis.common.mapper.{RedisCommand, RedisCommandDescription, RedisMapper}

object FlinkTest_12_Sink_Redis {


  def main(args: Array[String]): Unit = {


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val socketStream: DataStream[String] = env.socketTextStream("127.0.0.1", 8889)


    val config: FlinkJedisPoolConfig = new FlinkJedisPoolConfig.Builder()
      .setDatabase(1)
      .setHost("127.0.0.1")
      .setPort(6379)
      .build()

    socketStream
      .map((_, 1))
      .keyBy(0)
      .sum(1)
      .addSink(new RedisSink(config, new RedisMapper[(String, Int)] {
        override def getCommandDescription: RedisCommandDescription = {

          new RedisCommandDescription(RedisCommand.HSET, "car_wc")

        }

        override def getKeyFromData(t: (String, Int)): String = {
          t._1
        }

        override def getValueFromData(t: (String, Int)): String = {
          t._2.toString
        }
      }))

    env.execute("redis test...")

  }

}
